package ds_industry_2025.ds.ds_07.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

/*
    4、根据dwd层的数据，请计算每个省份累计订单量（订单信息表一条算一个记录），然后根据每个省份订单量从高到低排列，将结果打印到控制
    台（使用spark中的show算子，同时需要显示列名），将执行结果复制并粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务
    序号下；
 */
object t7 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("T2")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._



      spark.table("dwd.dim_province")
        .where("etl_date=(select max(etl_date) from dwd.dim_province)")
        .createOrReplaceTempView("p")

    spark.table("dwd.fact_order_info")
      .createOrReplaceTempView("o")

    val r1 = spark.sql(
      """
        |select distinct
        |p.name as province_name,
        |count(*) over(partition by o.province_id,p.name) as total_count
        |from o
        |join p on p.id=o.province_id
        |""".stripMargin)
      .orderBy(desc("total_count"))
      .withColumn("row",row_number().over(Window.orderBy(desc("total_count"))))

    val provinces = r1.select("province_name")
      .map(_(0).toString)
      .collect()

    r1.show

    //  todo 使用了pivot函数之后会把指定的列名取唯一值变成列名，并且会把其他字段的值也变成当前列的范围里面
    //  todo 然后在当前范围里面可以使用聚合函数对其他字段的值进行操作，然后作为当前列名的值
    val result = r1.groupBy()
      .pivot("province_name", provinces)
      .agg(first("total_count"))

    result.show



    result.show





    spark.close()
  }

}
